package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"log"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/go-redis/redis/v8"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"

	"github.com/tylertreat/BoomFilters"
)

// 配置常量（建议通过环境变量配置）
const (
	mysqlDSN    = "root:123456@tcp(localhost:3306)/ad_system?charset=utf8mb4&parseTime=True&loc=Local"
	redisAddr   = "localhost:6379"
	redisPass   = "123456"
	concurrency = 50 // 处理文件并发数  即每次并发处理多少行数据
)

// 数据结构定义
type (
	DeviceData struct {
		GAID      string `json:"gaid"`
		Geo       string `json:"geo"`
		OS        string `json:"os"`
		Publisher string `json:"publisher"`
		Bundle    string `json:"bundle"`
		Brand     string `json:"brand"`
		OSVersion string `json:"os_version"`
		Timestamp int64  `json:"timestamp"`
	}

	StrategyGroup struct {
		ID     uint64 `gorm:"primarykey"`
		AppID  string `gorm:"uniqueIndex"`
		Status string `gorm:"index"`
	}

	Strategy struct {
		ID         uint64         `gorm:"primarykey"`
		GroupID    uint64         `gorm:"index"`
		Priority   uint64         `gorm:"index"`
		Conditions map[string]any `gorm:"serializer:json"`
		Ratio      float64
	}

	OfferStrategy struct {
		ID         uint64 `gorm:"primarykey"`
		OfferId    uint64 `gorm:"index"`
		StrategyId uint64 `gorm:"index"`
	}

	Offer struct {
		ID           uint64 `gorm:"primarykey"`
		OfferID      string `gorm:"uniqueIndex"`
		Geo          string `gorm:"index"`
		OS           string `gorm:"index"`
		AppId        string `gorm:"index"`
		TrackingLink string
		DailyCap     int
		Status       string `gorm:"index"`
	}

	Audience struct {
		ID        uint64    `gorm:"primarykey"`
		S3Bucket  string    `gorm:"index"`
		Prefix    string    `gorm:"index"`
		Geo       string    `gorm:"index"`
		OS        string    `gorm:"index"`
		StartTime time.Time `gorm:"index"`
		EndTime   time.Time `gorm:"index"`
		Status    string    `gorm:"index"`
	}
)

type Processor struct {
	db         *gorm.DB
	rdb        *redis.Client
	s3Client   *s3.Client          // 替换s3svc
	downloader *manager.Downloader // 替换s3manager.Downloader

	// key为appId value为策略组
	appIdStrategyGroupMap map[string]StrategyGroup
	// key为offerId value为对应的策略数组
	offerStrategies map[uint64][]Strategy
	// key为appId value为对应的Offer数组
	appIdOffers map[string][]Offer

	// 新增：带时间窗口的布隆过滤器分片管理
	deviceShards   map[string]*boom.ScalableBloomFilter // 分片ID → 布隆过滤器
	shardLock      sync.RWMutex                         // 分片操作锁
	windowDuration time.Duration                        // 时间窗口（如2小时）
	shardInterval  time.Duration                        // 分片间隔（如1小时）
}

func main() {
	// 初始化处理器
	p := InitializeProcessor()
	defer p.Cleanup()

	// 启动人群包处理协程
	go p.ProcessAudiences()

	// 保持主程序运行
	select {}
}

func InitializeProcessor() *Processor {
	// 初始化数据库
	db, err := gorm.Open(mysql.Open(mysqlDSN), &gorm.Config{})
	if err != nil {
		log.Fatal("Database connection failed:", err)
	}

	// 自动迁移表结构
	if err := db.AutoMigrate(&StrategyGroup{}, &Strategy{}, &Offer{}, &Audience{}); err != nil {
		log.Fatal("Migration failed:", err)
	}

	// 初始化Redis
	rdb := redis.NewClient(&redis.Options{
		Addr:     redisAddr,
		Password: redisPass,
		DB:       0,
	})

	// 测试Redis连接
	if _, err := rdb.Ping(context.Background()).Result(); err != nil {
		log.Fatal("Redis connection failed:", err)
	}

	// 初始化AWS
	awsCfg, err := config.LoadDefaultConfig(context.TODO())
	if err != nil {
		log.Fatalf("AWS config error: %v", err)
	}
	s3Client := s3.NewFromConfig(awsCfg)

	p := &Processor{
		db:             db,
		rdb:            rdb,
		s3Client:       s3Client,
		downloader:     manager.NewDownloader(s3Client),
		windowDuration: 2 * time.Hour, // 2小时时间窗口
		shardInterval:  1 * time.Hour, // 每小时生成新分片
		deviceShards:   make(map[string]*boom.ScalableBloomFilter),
	}

	p.loadCaches()
	go p.startCacheRefresh()

	return p
}

// 启动分片清理
func (p *Processor) startShardCleaner() {
	ticker := time.NewTicker(p.shardInterval)
	for {
		select {
		case <-ticker.C:
			p.cleanupExpiredShards()
		}
	}
}

func (p *Processor) currentShardID() int {
	return int(time.Now().Unix() / int64(p.shardInterval.Seconds()))
}

// 清理过期的分片
func (p *Processor) cleanupExpiredShards() {
	p.shardLock.Lock()
	defer p.shardLock.Unlock()

	now := time.Now().Unix()
	cutoff := now - int64(p.windowDuration.Seconds())

	for shardID := range p.deviceShards {
		shardIDArr := strings.Split(shardID, "_")
		sid := shardIDArr[1]
		sidInt, err := strconv.ParseInt(sid, 10, 64)
		if err != nil {
			// 不合法的分片键
			delete(p.deviceShards, shardID)
		}
		shardTime := sidInt * int64(p.shardInterval.Seconds())
		if shardTime < cutoff {
			delete(p.deviceShards, shardID)
		}
	}
}

func (p *Processor) startCacheRefresh() {
	ticker := time.NewTicker(10 * time.Minute)
	for {
		select {
		case <-ticker.C:
			p.refreshCaches()
		}
	}
}

func (p *Processor) refreshCaches() {
	newSG := make(map[string]StrategyGroup)
	newOfferStrats := make(map[uint64][]Strategy)
	newAppIdOffers := make(map[string][]Offer)

	var sgs []StrategyGroup
	p.db.Find(&sgs)
	for _, sg := range sgs {
		newSG[sg.AppID] = sg
	}

	var strats []Strategy
	p.db.Find(&strats)
	stratsMap := make(map[uint64]Strategy)

	for _, s := range strats {
		stratsMap[s.ID] = s
	}
	var offerStrats []OfferStrategy
	p.db.Find(&offerStrats)
	for _, s := range offerStrats {
		newOfferStrats[s.OfferId] = append(newOfferStrats[s.OfferId], stratsMap[s.StrategyId])
	}

	// 重新加载Offer
	var offers []Offer
	p.db.Where("status = ?", "active").Find(&offers)
	for _, offer := range offers {
		newAppIdOffers[offer.AppId] = append(newAppIdOffers[offer.AppId], offer)
	}

	p.appIdStrategyGroupMap = newSG
	p.offerStrategies = newOfferStrats
	p.appIdOffers = newAppIdOffers
}

func (p *Processor) loadCaches() {
	// 加载策略组
	var sgs []StrategyGroup
	p.db.Find(&sgs)
	for _, sg := range sgs {
		p.appIdStrategyGroupMap[sg.AppID] = sg
	}

	// 加载策略
	var strats []Strategy
	p.db.Find(&strats)
	stratsMap := make(map[uint64]Strategy)

	for _, s := range strats {
		stratsMap[s.ID] = s
	}
	var offerStrats []OfferStrategy
	p.db.Find(&offerStrats)
	for _, s := range offerStrats {
		p.offerStrategies[s.OfferId] = append(p.offerStrategies[s.OfferId], stratsMap[s.StrategyId])
	}

	// 加载Offer
	var offers []Offer
	p.db.Where("status = ?", "active").Find(&offers)
	for _, offer := range offers {
		p.appIdOffers[offer.AppId] = append(p.appIdOffers[offer.AppId], offer)
	}
}

func (p *Processor) Cleanup() {
	if sqlDB, err := p.db.DB(); err == nil {
		err := sqlDB.Close()
		if err != nil {
			log.Printf("Error closing database: %v", err)
			return
		}
	}
	err := p.rdb.Close()
	if err != nil {
		log.Printf("Error closing database: %v", err)
		return
	}
}

func (p *Processor) ProcessAudiences() {
	for {
		// 获取所有激活的人群包
		var audiences []Audience
		if err := p.db.Where("status = 'active'").Find(&audiences).Error; err != nil {
			log.Println("Error fetching audiences:", err)
			// 简单重试
			time.Sleep(10 * time.Second)
			continue
		}

		// 并发处理每个人群包
		var wg sync.WaitGroup
		for _, audience := range audiences {

			var audienceAppIds []string

			if err := p.db.Table("offers").
				Select("distinct app_id").
				Where("status = ? AND geo= ? AND os = ?", "active", audience.Geo, audience.OS).
				Find(&audienceAppIds).Error; err != nil {
				log.Println("Error fetching audience app ids:", err)
				// 简单重试
				time.Sleep(10 * time.Second)
				continue
			}
			for _, appId := range audienceAppIds {
				wg.Add(1)
				go func(audience Audience, appId string) {
					defer wg.Done()
					p.ProcessAudience(audience, appId)
				}(audience, appId)
			}

		}
		wg.Wait()

		time.Sleep(30 * time.Second)
	}
}

func (p *Processor) ProcessAudience(audience Audience, appId string) {
	// 无限循环处理（按用户需求重复循环）
	for {
		// 从 endTime 开始，按小时倒序遍历到 startTime
		currentTime := audience.EndTime
		for currentTime.After(audience.StartTime) {
			// 获取当前小时的时间对象（截断到整点）
			currentHour := currentTime.Truncate(time.Hour)
			day := currentHour.Format("2006-01-02")
			hour := currentHour.Format("15") // 24小时制（如 "12"）

			// 遍历三个 region（顺序：eu → sg → us）
			for _, region := range []string{"eu", "sg", "us"} {
				files, err := p.ListS3Files(audience, day, hour, region)
				if err != nil {
					log.Printf("Failed to list files for %s %s %s: %v", day, hour, region, err)
					continue
				}

				if len(files) == 0 {
					continue
				}

				// 处理该小时和 region 的所有文件
				for _, file := range files {
					p.ProcessFile(audience.S3Bucket, *file.Key, appId)
				}
			}

			// 移动到上一小时
			currentTime = currentHour.Add(-time.Hour)
		}

		// 等待一段时间后重新开始循环（避免 CPU 占用过高）
		log.Printf("Finished processing all files for audience %d, restarting...", audience.ID)
		time.Sleep(30 * time.Second)
	}
}

// ListS3Files 根据 day、hour、region 参数列出 S3 中的文件
func (p *Processor) ListS3Files(audience Audience, day string, hour string, region string) ([]types.Object, error) {
	// 构造前缀
	basePrefix := audience.Prefix
	if basePrefix != "" && !strings.HasSuffix(basePrefix, "/") {
		basePrefix += "/"
	}
	fullPrefix := fmt.Sprintf(
		"%sday/%s/%s/%s-%s.%s.%s.%s/",
		basePrefix,
		audience.Geo,
		audience.OS,
		day,
		hour,
		audience.Geo,
		audience.OS,
		region,
	)

	input := &s3.ListObjectsV2Input{
		Bucket: aws.String(audience.S3Bucket),
		Prefix: aws.String(fullPrefix),
	}

	var allObjects []types.Object
	for {
		resp, err := p.s3Client.ListObjectsV2(context.TODO(), input)
		if err != nil {
			return nil, fmt.Errorf("failed to list objects: %v", err)
		}
		allObjects = append(allObjects, resp.Contents...)

		if !boolPtrToBool(resp.IsTruncated) {
			break
		}
		input.ContinuationToken = resp.NextContinuationToken
	}

	return allObjects, nil
}

// 辅助函数：将指针类型转换为 bool
func boolPtrToBool(p *bool) bool {
	if p == nil {
		return false
	}
	return *p
}

func (p *Processor) ProcessFile(bucketName string, key string, appId string) {
	// 下载文件
	// 创建带缓冲的写入器
	buf := manager.NewWriteAtBuffer([]byte{})

	_, err := p.downloader.Download(context.TODO(), buf, &s3.GetObjectInput{
		Bucket: aws.String(bucketName),
		Key:    aws.String(key),
	})

	if err != nil {
		log.Printf("Failed to download %s: %v", key, err)
		return
	}

	// 并行处理设备数据
	lines := strings.Split(string(buf.Bytes()), "\n")
	var wg sync.WaitGroup
	ch := make(chan string, 1000)

	// 启动worker池
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go p.deviceWorker(&wg, ch, appId)
	}

	// 分发任务
	for _, line := range lines {
		ch <- line
	}
	close(ch)
	wg.Wait()
}

func (p *Processor) deviceWorker(wg *sync.WaitGroup, ch <-chan string, appId string) {
	defer wg.Done()

	for line := range ch {
		var device DeviceData
		if err := json.Unmarshal([]byte(line), &device); err != nil {
			continue
		}
		p.ProcessDevice(device, appId)
	}
}

func (p *Processor) ProcessDevice(device DeviceData, appId string) {
	// 获取相关策略组
	group := p.appIdStrategyGroupMap[appId]

	groupIdStr := strconv.FormatUint(group.ID, 10)
	// 设备去重检查
	currentShardId := p.currentShardID()
	currentShard := fmt.Sprintf("%s_%d", groupIdStr, currentShardId)

	p.shardLock.RLock()
	bloomFilter, exists := p.deviceShards[currentShard]
	p.shardLock.RUnlock()

	if !exists {
		p.shardLock.Lock()
		defer p.shardLock.Unlock()
		if bloomFilter, exists = p.deviceShards[currentShard]; !exists {
			// 创建可扩展布隆过滤器（误判率0.1%，初始容量10000）
			bloomFilter = boom.NewDefaultScalableBloomFilter(0.01)
			p.deviceShards[currentShard] = bloomFilter
		}
	}

	// 获取关联的Offer
	offers := p.appIdOffers[appId]

	for _, offer := range offers {
		// 检查配额
		remaining := p.GetRemainingQuota(offer.ID)
		if remaining <= 0 {
			continue
		}

		// 匹配策略
		if p.MatchStrategies(offer, device) {
			p.SendClick(offer, device)
			break // 每个策略组只匹配一个offer
		}
	}
}

func (p *Processor) GetRemainingQuota(offerID uint64) int {
	val, err := p.rdb.Get(context.Background(), fmt.Sprintf("quota:%d", offerID)).Int()
	if errors.Is(err, redis.Nil) { // 初始化配额
		var offer Offer
		if err := p.db.First(&offer, offerID).Error; err != nil {
			return 0
		}
		p.rdb.Set(context.Background(), fmt.Sprintf("quota:%d", offerID), offer.DailyCap, 24*time.Hour)
		return offer.DailyCap
	}
	return val
}

func (p *Processor) MatchStrategies(offer Offer, device DeviceData) bool {
	strategies := p.offerStrategies[offer.ID]

	for _, s := range strategies {
		if p.MatchStrategy(s, device) {
			return true
		}
	}
	return false
}

func (p *Processor) MatchStrategy(s Strategy, d DeviceData) bool {
	// 检查版本要求
	if minVer, ok := s.Conditions["min_version"].(string); ok {
		if CompareVersion(d.OSVersion, minVer) < 0 {
			return false
		}
	}

	// 检查字段匹配
	for field, value := range s.Conditions {
		switch field {
		case "publisher":
			if d.Publisher != value {
				return false
			}
		case "bundle":
			if d.Bundle != value {
				return false
			}
		case "brand":
			if value != "" && value != d.Brand {
				return false
			}
		}
	}
	return true
}

func CompareVersion(v1, v2 string) int {
	parts1 := strings.Split(v1, ".")
	parts2 := strings.Split(v2, ".")

	for i := 0; i < len(parts1) || i < len(parts2); i++ {
		var n1, n2 int
		if i < len(parts1) {
			_, err := fmt.Sscanf(parts1[i], "%d", &n1)
			if err != nil {
				return 0
			}
		}
		if i < len(parts2) {
			_, err := fmt.Sscanf(parts2[i], "%d", &n2)
			if err != nil {
				return 0
			}
		}
		if n1 > n2 {
			return 1
		}
		if n1 < n2 {
			return -1
		}
	}
	return 0
}

func (p *Processor) SendClick(offer Offer, device DeviceData) {
	log.Printf("Sending click to %s | Device: %s", offer.TrackingLink, device.GAID)
}
